package com.rainbow.support.task.parseq;
import com.linkedin.parseq.Engine;
import com.linkedin.parseq.Task;
import com.linkedin.parseq.TaskType;
import com.linkedin.parseq.promise.Promises;
import com.linkedin.parseq.promise.SettablePromise;
import com.linkedin.parseq.trace.ResultType;
import com.linkedin.parseq.trace.ShallowTrace;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Supplier;


/**
 * 异步处理框架
 * Task的执行者代理类
 *        获取EngineAgent可以采用两种方式:
 *        采用单例EngineFactory方式
 *         EngineAgent engine=EngineFactory.getInstance().defaultEngine();
 * @author 傅为地
 */
@Slf4j
public class EngineAgent {

    private Engine engine;
    /**
     *线程池
     */
    private ExecutorService executors;

    /**
     *任务调度池程
     */
    private ScheduledExecutorService scheduler;

    /**
     * 是否单例线程池,默认采用enginFactory工厂单例模式
     * 1.通过spring容器时，线程池单例，关闭时，线程池不需要关闭
     * 2.通过enginFactory时，生成的引擎，执行完毕时需关闭线程池
     */

    public EngineAgent(Engine engine, ExecutorService executors, ScheduledExecutorService scheduler) {
        this.engine = engine;
        this.executors = executors;
        this.scheduler = scheduler;
    }

    /**
     * 开启异步执行
     * @param supplier  执行器
     * @param <T> 泛型类
     * @return 执行结果
     */
    public <T> SettablePromise<T> async(Supplier<T> supplier) {
        final SettablePromise<T> promise = Promises.settable();
        executors.execute(() -> {
            try {
                promise.done(supplier.get());
            } catch (Exception e) {
                promise.fail(e);
                log.error("ParSeq异步执行,异常:{}",e);
            }
        });
        return promise;
    }

    public <T> Task<T> task(Supplier<T> supplier) {
        return Task.async(() -> async(supplier));
    }

    /**
     * 无参任务执行
     * @param task
     */
    public void run(final Task<?> task) {
        engine.run(task);
    }

    /**
     * 关闭引擎
     */
    public void shutdown() {
        engine.shutdown();
        scheduler.shutdown();
    }

    /**
     * 获取引擎
     * @return 返回执行引擎
     */
    public Engine getEngine() {
        return engine;
    }

    /**
     * 多个异步任务执行出现异常时，获取异步任务列表中出现异常任务的堆栈信息
     *
     * @param task 执行的任务
     * @return
     */
    public Set<ShallowTrace> getParseqTaskErrors(Task task) {
        Set<ShallowTrace> errors = new LinkedHashSet<>();
        Map<Long, ShallowTrace> traceMap = task.getTrace().getTraceMap();
        if (ObjectUtils.isNotEmpty(traceMap)) {
            traceMap.entrySet().stream()
                    .map(Map.Entry::getValue)
                    .filter(trace -> ObjectUtils.isNotEmpty(trace))
                    .filter(trace -> Objects.equals(ResultType.ERROR, trace.getResultType()))
                    .filter(trace -> StringUtils.equals(TaskType.FUSION.getName(), trace.getTaskType()))
                    .forEach(trace -> errors.add(trace));
        }
        return errors;
    }

    /**
     * 多个异步任务执行出现异常时，打印异步任务列表中出现异常任务的堆栈信息
     * @param task 执行的任务
     */
    public void printParseqTaskErrors(Task task) {
        Map<Long, ShallowTrace> traceMap = task.getTrace().getTraceMap();
        if (ObjectUtils.isNotEmpty(traceMap)) {
            traceMap.entrySet().stream()
                    .map(Map.Entry::getValue)
                    .filter(trace -> ObjectUtils.isNotEmpty(trace))
                    .filter(trace -> Objects.equals(ResultType.ERROR, trace.getResultType()))
                    .filter(trace -> StringUtils.equals(TaskType.FUSION.getName(), trace.getTaskType()))
                    .forEach(trace -> log.error(
                            "ParSeq异步任务名称=>:{},异常信息=>:{}",
                            trace.getName(),
                            trace.getValue()));
        }
    }
}
